Python sqlalchemy

sqlalchemy 是 python 的 ORM(Object-relational mapping 对象关系映射)框架,
通过对象和数据库表之间进行映射，从而实现对象对数据库的操作.

安装

 $ pip install SQLAlchemy
 $ pip list | grep SQLAlchemy
 > SQLAlchemy           2.0.23

连接

sqlalchemy 会根据数据库驱动自动创建数据库文件并连接数据库

from sqlalchemy import create_engine

engine = create_engine(
    "mysql://user:password@localhost:3306/dbname",
    echo=True,                                   # True 打印出 sql 语句, 方便调试
    future=True,                                 # 使用 SQLAlchemy 2.0 API，向后兼容
    pool_size=5,                                 # 连接池大小默认为 5 个，设置 0 表示连接无限制
    pool_recycle=3600,                           # 设置时间, 限制数据库自动断开
)

engine = create_engine(
    "sqlite:////root/Desktop/sqilte.db",         # 创建 SQLite 的内存数据库, 路径使用绝对路径
    echo=True, future=True,                      # echo=True, 打印执行的 sql 语句
    connect_args={"check_same_thread": False}    # 须加上 check_same_thread=False，否则多线程中无法使用
)

创建表

sqlalchemy 内置 declarative_base 用于创建表的基类,
所有继承该类的类都会在注册到内存后自动创建表

创建 models.py 文件, 继承 Base 类, 定义表模型

from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()                        # 声明表的基类, 所有表继承 Base

class User(Base):
    __tablename__ = 'users'                      # 表名

    id = Column(Integer, primary_key=True)       # 设置主键
    name = Column(String)
    age = Column(Integer)

    def __repr__(self):                          # 打印出表详细信息
        return f"<User(id={self.id}, name={self.name}, age={self.age})>"

注: 更多表数据类型数据类型

创建 sql.py 文件, 进行数据库连接并创建表

from sqlalchemy import create_engine

from models import Base                          # 引入表模型, 将表模型注册进内存

engine = create_engine(
    "sqlite:////root/Desktop/sqilte.db",
    echo=True, future=True,
    connect_args={"check_same_thread": False}
)
Base.metadata.create_all(engine, checkfirst=True)# 所有表注册进内存后, 创建表

在 user.py 使用数据库

from sqlalchemy.orm import Session

from sql import engine
from models import User

db = Session(engine)                             # 创建数据库会话
db.query(User).all()                             # 数据库操作
db.close()                                       # 关闭数据库会话

插入数据

sqlalchemy 1.x 使用 db.add 添加数据
sqlalchemy 2.x 使用 insert 生成 sql 语句, 然后使用 db.execute 执行

from sqlalchemy import insert
from sqlalchemy.orm import declarative_base, sessionmaker, Session

db = Session(engine)                                            # 创建 Session 对象

db.add(User(name="Jason", age=20))                                   # sqlalchemy 1.x 插入单个数据
db.add_all([                                                         # sqlalchemy 1.x 插入多个数据
    User(name="Jason", age=20),
    User(name="Lily", age=18),
])
db.execute(insert(User), [                                           # sqlalchemy 2.x 插入数据
    {'name':"Jason", 'age':20},
    {'name':"Lily", 'age':18},
])

db.commit()                                                          # 增删改查之后都需要提交数据, 令修改生效
db.close()

stmt = insert(User).values(name=Jason, age=20)                       # Insert.values 添加单个数据
stmt = insert(User).values([{'name':"Jason", 'age':20}])             # Insert.values 批量添加数据

stmt = insert(User).values(name=Jason, age=20).returning(User)       # Insert.returning, execute 时返回插入数据
db.execute(stmt)                                                     # 执行 sql 语句

查询数据

sqlalchemy 1.x 使用 db.query 查询数据

from sqlalchemy import select

db.query(User).filter(User.age == 18).first()                        # sqlalchemy 1.x 获取单个数据
db.query(User).filter(User.age == 18).all()                          # sqlalchemy 1.x 批量获取数据

stmt = select(User).where(User.age == 18)
db.execute(stmt).scalar()                                            # sqlalchemy 2.x 获取单个数据
db.execute(stmt).scalars()                                           # sqlalchemy 2.x 批量获取数据

db.commit()

stmt = select(User).where(User.age == 18).order_by(User.id)          # 查询数据并按 id 排序
db.execute(stmt)
db.commit()

更新数据

from sqlalchemy import update

user = db.query(User).filter(User.age == 18).first()                 # 通过 query 筛选出单个数据
user.age = 19                                                        # sqlalchemy 1.x 更新单个数据
db.query(User).filter(User.age == 18).update({"age": 19})            # sqlalchemy 1.x 批量更新

db.execute(update(User).where(User.age == 18), {"age": 19})          # sqlalchemy 2.x 批量更新
db.execute(update(User).where(User.age == 18).values(age=19))

db.commit()
db.close()

删除数据

from sqlalchemy import delete

db.delete(db.query(User).filter(User.age == 18).first())             # sqlalchemy 1.x 删除单个数据
db.query(User).filter(User.age == 18).delete()                       # sqlalchemy 1.x 批量删除数据

db.execute(delete(User).where(User.age == 18))                       # sqlalchemy 2.x 批量删除数据

db.commit()
db.close()

拓展

表数据类型

Integer     : int 整数类型
Float       : float 浮点类
Boolean     : bool 布尔类型, true, false
String      : string 字符和字符串的基类
Time        : datetime.time() 类型对象

SmallInteger: 较小的一种 int 整数
BigInteger  : 一种更大的 int 整数。
Double      : 一种更大的 float 浮点数

Date        : datetime.datetime() 类型对象
DateTime    : datetime.datetime() 类型对象
Interval    : datetime.timedelta() 类型对象

Text        : 大小可变的字符串类型。
Unicode     : 长度可变的Unicode字符串类型
UnicodeText : 无限长的Unicode字符串类型

Enum        : 枚举类型
ARRAY       : 数组类型 Column("data", ARRAY(Integer)) => {"data": [1,2,3]}
JSON        : 字典类型 Column('data', JSON) => {"data": {"a": 1, "b": 2}}

表模块封装

创建一个基类表, 并赋予一些基本功能
便于表数据与基本数据的转换与变更

class Table(Base):
    __abstract__ = True

    id = Column(Integer, primary_key=True)

    def __init__(self):
        pass

    def get(self, key):
        """ 获取对象属性 """
        return getattr(self, key)

    def set(self, key, value):
        """ 设置对象属性 """
        return setattr(self, key, value)

    def update(self, items):
        """ 更新对象属性 """
        for key, value in items.items():
            setattr(self, key, value)

    def json(self):
        """ 对象转为 json """
        return {col.name: getattr(self, col.name, None) for col in self.__table__.columns}

    def __repr__(self):
        return f'<Table id:{self.id}>'

class User(Table):
    """ 继承表基类, 创建用户表 """

    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

    def __init__(self, name, age):
        super().__init__()
        self.name = name

    def __repr__(self):
        return f"<User(id={self.id}, name={self.name}, age={self.age})>"